package com.ruoyi.system.config;


import com.ruoyi.system.constant.MqttPushClient;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

@Component
public class PushCallback implements MqttCallbackExtended {
    private static final Logger logger = LoggerFactory.getLogger(PushCallback.class);
    @Autowired
    private MqttPushClient mqttPushClient;

    @Autowired
    private MqttConfig mqttConfig;

    private String _msg;

    /**
     * 丢失连接，可在这里做重连
     * 只会调用一次
     *
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        // 连接丢失后，一般在这里面进行重连
        logger.info("===>>>" + mqttConfig.getInClientId() + "与MQTT连接断开，发起重连...");
        if (MqttPushClient.getClient() == null || !MqttPushClient.getClient().isConnected()) {
            mqttConfig.getMqttPushClient();
        }
    }

    /**
     * subscribe后得到的消息会执行到这里面
     * 消息到达的回调
     *
     * @param topic
     * @param mqttMessage
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) {
        _msg = new String(mqttMessage.getPayload(), StandardCharsets.UTF_8);
        System.out.println("_msg: " + _msg);

//        try {
//            MqttDispose mqttDataService = SpringUtils.getBean(MqttDispose.class);
//            mqttDataService.disposeBusiness(topic, _msg);
//
//        } catch (Exception e) {
//            logger.info("处理订阅消息异常: " + e.getMessage(), e);
//        }
    }

    /**
     * publish后，配送完成后回调的方法
     *
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        if (iMqttDeliveryToken.isComplete()) {
            logger.info("publish消息成功,消息ID: {}" , iMqttDeliveryToken.getMessageId());
        } else {
            logger.info("publish消息失败,原因: {}" , iMqttDeliveryToken.getException().getMessage());
        }
    }

    /**
     * 连接成功后的回调执行订阅主题
     * 重新连接后主题也需要再次订阅,将重新订阅主题放在连接成功后的回调
     *
     * @param reconnect
     * @param serverURI
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        logger.info("===>>>MQTT连接成功<<<===,连接方式：{}", reconnect ? "重连" : "直连");
        try {
            if (MqttPushClient.getClient().isConnected()) {
                //连接成功后订阅主题
                for (String topic : mqttConfig.getTopicList()) {
                    mqttPushClient.cleanTopic(topic);
                    mqttPushClient.subscribe(topic, 0);
                }

                logger.info("mqtt连接成功，客户端ID：" + mqttConfig.getInClientId());
            } else {
                logger.info("mqtt连接失败，客户端ID：" + mqttConfig.getInClientId());
            }
        } catch (Exception e) {
            logger.info("mqtt订阅主题异常:" + e);
        }
    }
}
